package com.rtf.framework.es.config;

import com.rtf.framework.common.util.UtilString;
import com.google.common.base.Splitter;
import com.google.common.collect.Lists;
import com.google.common.primitives.Ints;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.client.DefaultConnectionKeepAliveStrategy;
import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.nio.reactor.IOReactorException;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;

import java.util.List;

@Slf4j
@Configuration
@EnableConfigurationProperties(AppElasticSearchProperties.class)
@ComponentScan(basePackages = "com.rtf.framework.es")
public class AppElasticSearchConfig {

	@Bean(destroyMethod = "close")
	public RestHighLevelClient restHighLevelClient(AppElasticSearchProperties esProp) {

		List<HttpHost> httpHosts = Lists.newArrayList();
		Splitter.on(",").withKeyValueSeparator(":").split(esProp.getHosts()).forEach((host, port) -> {
			httpHosts.add(new HttpHost(host, Ints.tryParse(port), esProp.getScheme()));
		});
		RestClientBuilder restClientBuilder = RestClient.builder(httpHosts.toArray(new HttpHost[] {}));

		restClientBuilder.setRequestConfigCallback(requestConfigBuilder ->
				requestConfigBuilder.setConnectTimeout(esProp.getConnectTimeout()).setSocketTimeout(esProp.getSocketTimeout()));

		restClientBuilder.setHttpClientConfigCallback(httpClientBuilder -> {

			PoolingNHttpClientConnectionManager connManager = null;
			try {
				connManager = new PoolingNHttpClientConnectionManager(new DefaultConnectingIOReactor(IOReactorConfig.DEFAULT));
				connManager.setMaxTotal(esProp.getMaxConnection());
				connManager.setDefaultMaxPerRoute(esProp.getMaxConnectionPerRoute());
			} catch (IOReactorException e) {
				log.error(e.getMessage());
			}
			httpClientBuilder.setConnectionManager(connManager);

			//显式启用 TCP keepalive 设置linux的net.ipv4.tcp_keepalive_time=300, @see: https://github.com/elastic/elasticsearch/issues/65213
			httpClientBuilder.setDefaultIOReactorConfig(IOReactorConfig.custom()
					.setSoKeepAlive(true)
					.build());

			/**
			 * 权限验证信息
			 */
			if (UtilString.isNotBlank(esProp.getUserName()) && UtilString.isNotBlank(esProp.getPassword())) {
				CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
				credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(esProp.getUserName(), esProp.getPassword()));
				httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
			}

			return httpClientBuilder;
		});

		return new RestHighLevelClient(restClientBuilder);

	}
}
